package com.shujia.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo02WordCountOnYarn {
  /**
   * 提交到集群的一般步骤：
   * 1、注释setMaster
   * 2、将路径换成HDFS的目录：
   *  hdfs dfs -mkdir -p /spark/data/wc
   * 3、将数据上传到对应的目录
   *  hdfs dfs -put words.txt /spark/data/wc
   * 4、使用maven将程序打成jar包，上传至虚拟机
   * 5、使用命名进行提交：
   *  spark-submit --class com.shujia.core.Demo02WordCountOnYarn --master yarn-client spark-1.0.jar
   *
   */
  def main(args: Array[String]): Unit = {
    // 创建Spark的环境
    val conf: SparkConf = new SparkConf() // 对Spark进行配置
    conf.setAppName("Demo02WordCountOnYarn") // 设置程序名称
    //    conf.setMaster("local") // 设置Spark的运行方式，如果是本地 则设为local即可

    val sc: SparkContext = new SparkContext(conf)

    // 1、读取数据
    /**
     * RDD：在Spark中最核心的东西
     * 弹性分布式数据集
     * 先把它当成List去使用
     * 统一了编程模型
     */
    val lineRDD: RDD[String] = sc.textFile("/spark/data/wc")

    // 2、将每一个单词切分出来
    val wordsRDD: RDD[String] = lineRDD.flatMap(_.split(","))

    // 3、按照每个单词进行分组
    val grpRDD: RDD[(String, Iterable[String])] = wordsRDD.groupBy(word => word)

    // 4、统计单词的数量
    val wordCntRDD: RDD[String] = grpRDD.map(kv => s"${kv._1},${kv._2.size}")

    // 5、在集群环境一般不直接打印数据，通常会进行保存
    wordCntRDD.saveAsTextFile("/spark/output/wc")


  }


}
